package com.deepglint.tableapi;

import com.deepglint.beans.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.Tumble;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @author mj
 * @version 1.0
 * @date 2021-11-28 15:10
 */
public class TableTest_TimeAndWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1.表的创建：连接外部系统，读取数据
        // 1.1 读取文件
        String path = "C:\\Users\\马军\\Desktop\\Idea-workspace\\flink\\src\\main\\resources\\source.txt";
        DataStreamSource<String> source = env.readTextFile(path);

        // 2. 转为pojo
        SingleOutputStreamOperator<SensorReading> mapStream = source.map(line -> {
            String[] split = line.split(",");
            return new SensorReading(split[0], split[1], new Long(split[2]), new Double(split[3]));
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(2)) {
            @Override
            public long extractTimestamp(SensorReading element) {
                return element.getTimestamp();
            }
        });

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 3. 将流转为表,并定义时间特性
//        Table table = tableEnv.fromDataStream(mapStream, "id,timestamp as ts,pt.proctime"); // proc time
        Table table = tableEnv.fromDataStream(mapStream, "id,timestamp as ts,temperature as temp,rt.rowtime"); // 事件时间

        // 4.注册表
        tableEnv.createTemporaryView("sensor", table);

        // 5 窗口操作
        // 5.1 group window
        // table API
        Table resultTable = table.window(Tumble.over("10.seconds").on("rt").as("tw"))
                .groupBy("id,tw")
                .select("id,id.count,temp.avg,tw.end");

        // 另外一种写法
        Table resultSqlTable = tableEnv.sqlQuery("select id,count(id) as cnt,avg(temp) as avgTemp,tumble_end(rt,interval '10' second) " +
                "from sensor group by id,tumble(rt,interval '10' second)");

        // 5.2 over window
        Table resultOverTable = table.window(Over.partitionBy("id").orderBy("rt").preceding("2.rows").as("ow"))
                .select("id,rt,id.count over ow,temp.avg over ow");

        // 另外一种写法，sql
        Table resultOverSqlTable = tableEnv.sqlQuery("select id,rt,count(id) over ow,avg(temp) over ow " +
                "from sensor " +
                "window ow as (partition by id order by rt rows between 2 preceding and current row)");


//        table.printSchema();

        tableEnv.toRetractStream(resultTable, Row.class).print("table");
        tableEnv.toRetractStream(resultSqlTable, Row.class).print("sql");
        tableEnv.toRetractStream(resultOverTable, Row.class).print("overTable");
        tableEnv.toRetractStream(resultOverSqlTable, Row.class).print("overSql");

        env.execute();
    }
}
